package com.poetic.order;

import com.poetic.order.domain.OrderEvent;
import com.poetic.order.domain.OrderResult;
import com.poetic.order.function.OrderPaySelect;
import com.poetic.order.function.OrderTimeOutSelect;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * <pre>
 *  们先将事件流按照订单号 orderId
 * 分流，然后定义这样的一个事件模式：在 15 分钟内，事件“create”与“pay”非严
 * 格紧邻
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class OrderTimeoutTask {
    public static void main(String[] args) throws Exception {
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/OrderLog.csv";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        KeyedStream<OrderEvent, Long> dataStream = env.readTextFile(path)
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String data) throws Exception {
                        String[] line = data.split("\\,");

                        return new OrderEvent(Long.valueOf(line[0]),
                                line[1],
                                line[2],
                                Long.valueOf(line[3]));
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getEventTime() * 1000;
                            }
                        })).keyBy(new KeySelector<OrderEvent, Long>() {
                    @Override
                    public Long getKey(OrderEvent value) throws Exception {
                        return value.getOrderId();
                    }
                });
        dataStream.print();
        // 定义一个匹配模式  
        Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("begin")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.getEventType().equals("create");
                    }
                })
                .followedBy("follow")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.getEventType().equals("pay");
                    }
                })
                .within(Time.minutes(15));
        // 把模式应用到stream上，得到一个pattern stream
        PatternStream<OrderEvent> patternStream = CEP.pattern(dataStream, orderPayPattern);
        // 调用select方法，提取事件序列，超时的事件要做报警提示
        OutputTag<OrderResult> orderTimeoutOutputTag = new OutputTag<OrderResult>("orderTimeout"){};

        SingleOutputStreamOperator<OrderResult> resultStream = patternStream.select(orderTimeoutOutputTag,
                new OrderTimeOutSelect(),
                new OrderPaySelect());

        resultStream.print("payed");
        resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout");
        env.execute("order timeout");
    }
}
